package tuijianhaoyou;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;


import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


import java.io.File;
import java.io.IOException;


public class HaoYou {
    public static class HaoYouMap extends Mapper<LongWritable, Text,Text,Text> {
         Text x = new Text();
         Text y = new Text();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] strings = value.toString().split("\t");
            String a = strings[0];
            String[] b = strings[1].split(" ");

            x.set(strings[0]);
            y.set(strings[1]);

            context.write(x,y);
        }
    }
    public static class HaoYouReduce extends Reducer<Text,Text,Text,Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            StringBuffer buffer = new StringBuffer();
            for (Text value : values){
                buffer.append(value + "," );
            }
            context.write(key, new Text(buffer.toString()));
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
      try {
          Job job = Job.getInstance(conf);

          job.setJarByClass(HaoYou.class);

          job.setMapperClass(HaoYouMap.class);
          job.setReducerClass(HaoYouReduce.class);

          job.setMapOutputKeyClass(Text.class);
          job.setMapOutputValueClass(Text.class);

          job.setOutputKeyClass(Text.class);
          job.setOutputKeyClass(Text.class);

          File file = new File("f:/all/a/tuijianhaoyou/xxx");
          if (file.exists()){
              FileUtil.fullyDelete(file);
          }

          FileInputFormat.addInputPath(job,new Path("f:/all/a/tuijianhaoyou/xx"));
          FileOutputFormat.setOutputPath(job,new Path("f:/all/a/tuijianhaoyou/xxx"));

          job.setNumReduceTasks(1);

          System.exit(job.waitForCompletion(true)? 0:1);
      } catch (Exception e){
          e.printStackTrace();
      }
    }
}
